import copy
from datetime import datetime
import time
import gc
import json
import pandas as pd
import socket
import struct
import uuid

from .u_exception import capt_exc, ErrorLogger
from .common_func.print_logger import PrintLogger

import_error_warned = False
try:
    import pandas as pd
except ModuleNotFoundError:
    if not import_error_warned:
        PrintLogger.warning("DataFrameFilter is not valid, because require module:[pandas] is not installed. "
                            "If you want use DataFrameFilter, please install [pandas] first")
        import_error_warned = True

error_subject_name = "UFilter"


def is_iterable_but_not_str(x):
    return hasattr(x, '__iter__') and not isinstance(x, str)


class Expression(object):
    def __init__(self, **kwargs):
        self.key = kwargs.get("key", None)
        self.value = kwargs.get("value", "")
        self.filter_value = self.value
        self.operator = kwargs.get("operator", None)

        if kwargs:
            if self.operator == Filter.Operations.LIKE:
                self.filter_value = "'%%%s%%'" % self.value
            elif self.operator == Filter.Operations.GLOB:
                self.filter_value = "'*%s*'" % self.value
            elif self.operator == Filter.Operations.IN:
                self.filter_value = "('%s')" % "','".join(self.value)
            else:
                if type(self.value) is bool:
                    self.filter_value = 1 if self.value else 0
                else:
                    self.filter_value = "'%s'" % self.value

    def filter_string(self):
        filter_str = "%s %s %s" % (self.key, self.operator, self.filter_value)
        return filter_str

    def filter(self):
        filter_str = "? %s ?" % self.operator
        filter_values = (self.key, self.operator, self.filter_value)
        return filter_str, filter_values

    def __str__(self):
        return self.filter_string()


class ExpressionGroup(object):

    def __init__(self):
        self.expression = None
        self.exp_group1 = None
        self.exp_group2 = None
        self.exp_operator = None

    def is_dummy(self):
        return self.expression is None and self.exp_operator is None

    def is_filtered(self):
        return not self.is_dummy()

    def __or__(self, exp_group):
        if exp_group.is_dummy() or self.is_dummy():
            return exp_group if self.is_dummy() else self
        new_exp_group = type(self)()
        new_exp_group.exp_group1 = exp_group
        new_exp_group.exp_group2 = self
        new_exp_group.exp_operator = Filter.Operations.OR
        return new_exp_group

    def __and__(self, exp_group):
        if exp_group.is_dummy() or self.is_dummy():
            return exp_group if self.is_dummy() else self
        new_exp_group = type(self)()
        new_exp_group.exp_group1 = exp_group
        new_exp_group.exp_group2 = self
        new_exp_group.exp_operator = Filter.Operations.AND
        return new_exp_group

    def __str__(self):
        if self.is_dummy():
            return ""
        if self.expression:
            return self.expression.filter_string()
        else:
            return "((%s) %s (%s))" % (self.exp_group1.filter_string(), self.exp_operator,
                                       self.exp_group2.filter_string())

    def filter(self):
        if self.is_dummy():
            return "", []
        if self.expression:
            return self.expression.filter()
        else:
            group1_str, group1_values = self.exp_group1.filter()
            group2_str, group2_values = self.exp_group2.filter()
            filter_str = "((%s) %s (%s))" % (group1_str, self.exp_operator, group2_str)
            return filter_str, group1_values + group2_values

    def filter_string(self):
        return str(self)


class Filter(object):
    """A filter to be used to filter the results from a database query.
    Users should not have to use this class."""

    class Operations:
        AND = 'AND'
        OR = 'OR'
        LIKE = 'LIKE'
        GLOB = 'GLOB'
        IN = 'IN'
        EQ = '='
        NE = '!='
        LT = '<'
        LE = '<='
        GT = '>'
        GE = '>='

    def __init__(self, buf, key):
        self.buf = buf
        self.key = key
        self.expression_group = ExpressionGroup()
        self.expression_t = Expression

    def is_filtered(self):
        """If the filter contains any filters"""
        return not self.expression_group.is_dummy()

    def _comparison(self, value, operation):
        self.expression_group.expression = self.expression_t(key=self.key,
                                                             value=value, operator=operation)
        return self

    def like(self, value):
        """Perform LIKE operation"""
        return self._comparison(value, self.Operations.LIKE)

    def glob(self, value):
        """Perform GLOB operation"""
        return self._comparison(value, self.Operations.GLOB)

    def __eq__(self, value):
        """Perform EQUALS operation
        When input value is an iterable, but not a string, it will match for
        any of the values on the iterable
        """
        # Iterable, so we use IN (X, X...) instead of =
        if is_iterable_but_not_str(value):
            return self._comparison(value, self.Operations.IN)
        else:
            return self._comparison(value, self.Operations.EQ)

    def __ne__(self, value):
        """Perform NOT EQUALS operation"""
        return self._comparison(value, self.Operations.NE)

    def __lt__(self, value):
        """Perform LESS THAN operation"""
        return self._comparison(value, self.Operations.LT)

    def __le__(self, value):
        """Perform LESS THAN OR EQUALS operation"""
        return self._comparison(value, self.Operations.LE)

    def __gt__(self, value):
        """Perform GREATER THAN operation"""
        return self._comparison(value, self.Operations.GT)

    def __ge__(self, value):
        """Perform GREATER THAN OR EQUALS operation"""
        return self._comparison(value, self.Operations.GE)

    def __and__(self, other_filter):
        """
        Returns a new filter that combines this filter with other_filter using AND.
        """
        new_filter = type(self)(self.buf, None)
        new_filter.expression_group = self.expression_group & other_filter.expression_group
        return new_filter

    def __or__(self, other_filter):
        """
        Returns a new filter that combines this filter with other_filter using OR.
        """
        new_filter = type(self)(self.buf, None)
        new_filter.expression_group = self.expression_group | other_filter.expression_group
        return new_filter

    def __len__(self):
        """Returns the number of records that matches this filter"""
        if self.expression_group.is_dummy():
            count = len(self.buf)
        else:
            count = self.buf.get_len(buf_filter=self.expression_group)
        return count

    def __iter__(self):
        """Returns in iterator over the records for this filter"""
        if self.expression_group.is_dummy():
            res = self.buf()
        else:
            res = self.buf(self.expression_group)
        return iter(res)

    def __str__(self):
        """Returns a string representation of the filter"""
        if self.expression_group:
            return self.expression_group.filter_string()
        else:
            return ""

    def filter(self):
        """Returns the filter"""
        if self.expression_group:
            return self.expression_group.filter()
        else:
            return "", []


# 以下仅用于接口，用以返回给人机接口告知操作人可选的操作，先放在这里，后续再调整结构
STR_OP = {'__add__': '加', '__len__': '字符串长度', '__mul__': '乘(N个该字符串拼接)', 'lower': "小写",
          'lstrip': "字符串头去除", 'replace': "内容替换", 'rstrip': "字符串尾去除", 'strip': "字符串头尾都去除",
          'upper': "大写", 'zfill': "头部用0填充到指定长度"}

JUDGE = {'isdigit': '是纯数字字符串', '__gt__': '大于', '__lt__': '小于', '__ge__': "大于等于", '__le__': "小于等于",
         'startswith': "以...开始", 'endswith': "以...结束", '__ne__': "不等于", '__contains__': "包含",
         '__eq__': "等于",
         'in': "在...中", 'between': "介于...之间(含边界值)", 'not_in': "不在...中",
         'not_between': "不介于...之间(不含边界值)"}

NUM_OP = {'__add__': '加', '__abs__': '绝对值', '__mul__': '乘', '__floordiv__': "除后取商", '__mod__': "除后取余",
          '__pow__': "冪次方", '__str__': "转字符串格式", '__sub__': "减"}

DATETIME_OP = {'__str__': '转字符串格式', 'day': '日', 'date': '日期', 'hour': '时', 'microsecond': '毫秒',
               'minute': '分', 'month': '月', 'second': '秒', 'weekday': '周几', 'year': '年'}
VALID_OP_SET = set(STR_OP.keys() | NUM_OP.keys() | DATETIME_OP.keys())
EXPR_KEYS = {"filter_value", "filter_op", "op_value1", "op_value2"}
EXPR_MUST_HAS_KEYS = {"filter_value", "filter_op", "op_value1"}
ASSET_NAME_SET = {"api", "app", "account", "host", "org"}


class DataFrameFilter:
    """
    变更说明：老版本的过滤器逻辑是将过滤条件解析成python表达式，并落入临时函数_func中，将数据框逐行执行apply(_func)来过滤数据
            这种过滤方式效率较低，经研究，如果通过数据框列向量计算的方式来实现，效率会提高100到1000倍
            本版本即改为尽可能的使用列向量计算的方式实现数据过滤
                所有的对字段进行变换的操作，均先富化为数据框的一个新列,这样才有可能做列向量化计算

    DataFrameFilter 类是过滤日志和数据转换操作的整合类，用于在Dataframe中根据过滤条件过滤出数据，返回符合条件的Dataframe。
    主入口是filter_data方法，其余方法均为此方法的支撑函数
    过滤条件格式为列表，形式为[过滤表达式, and或or连接符， 过滤表达式1, and或or连接符， 过滤表达式2...]
    过滤表达式格式为字典，形式为{filter_value：数值表达式，filter_op: 过滤判断方法字符串， op_value1：数值表达式， op_value2：数值表达式}
    过滤判断方法支持的方法有：
        {'isdigit': '是纯数字字符串', '__gt__': '大于', '__lt__': '小于', '__ge__': "大于等于", '__le__': "小于等于",
         'startswith': "以...开始", 'endswith': "以...结束", '__ne__': "不等于", '__contains__': "包含", '__eq__': "等于",
         'in': "在...中", 'between': "介于...之间", 'not_in': "在...中", 'not_between': "介于...之间"}
    数值表达式格式为字典， 形式为{const_value: 固定值， \
                            'asset': 资产表名（"api":/"app"/"account"/"host"), \
                            field_name: log_hour, trans: 转换方法字符串, args: 转换参数}
    转换方法字符串支持的方法有：
        STR_OP_LIST = {'__add__': '加', '__len__': '字符串长度', '__mul__': '乘(N个该字符串拼接)', 'lower': "小写",
          'lstrip': "字符串头去除", 'replace': "内容替换", 'rstrip': "字符串尾去除", 'strip': "字符串头尾都去除",
          'upper': "大写", 'zfill': "头部用0填充到指定长度"}
        NUM_OP_LIST = {'__add__': '加', '__abs__': '绝对值', '__mul__': '乘', '__floordiv__': "除后取商", '__mod__': "除后取余",
          '__pow__': "冪次方", '__str__': "转字符串格式", '__sub__': "减"}
        DATETIME_OP_LIST = {'__str__': '转字符串格式', 'date': '日期', 'day': '日', 'hour': '时', 'microsecond': '毫秒',
                    'minute': '分', 'month': '月', 'second': '秒','weekday': '周几', 'year': '年'}
    转换参数可以是数值或字符串，多个字符串以|连接作为一个参数

    For Example：
        过滤条件:
        filter_condition = [{"filter_value"：{"field_name": logtime, "trans": "hour", args: []}，
                             "filter_op": "between"，
                             "op_value1"：{"const_value": 8}，
                             "op_value2"：{"const_value": 20}}]
        这是一个单过滤条件
        意思为：取Dataframe的logtime字段，并对其执行hour方法，即：logtime的hour结果在固定值8和固定值20之间，
            业务含义为过滤工作时间的日志，工作时间设置为8点到20点

    """

    def __init__(self, asset_cls_obj=None, logger=PrintLogger):
        self.logger = logger
        self._method_list = list()
        self._op_dict = dict()
        self._special_judge_list = ['in', 'between', 'not_in', 'not_between']
        ErrorLogger.add_logger(error_subject_name, logger)
        if not hasattr(asset_cls_obj, "get_ast"):
            self.logger.warning("没指定资产类或资产类没有 get_ast 方法， 从资产中过滤数据的功能不可用")
            self.asset_cls = None
            self.asset_name_set = set()
        else:
            self.asset_cls = asset_cls_obj
            try:
                name_map = asset_cls_obj.get_ast()
                self.asset_name_set = set(name_map.keys())
            except Exception as e:
                self.logger.debug(e)
                self.asset_name_set = ASSET_NAME_SET

    def get_filter_method_list(self):
        """
        获取过滤操作的方法列表，返回元素为方法描述字典的列表
        方法描述字典格式为 {"method_name": method_name, "display_name": display_name}
        如：{"method_name": "between", "display_name": "介于...之间"}
        :return:
        """
        if not self._method_list:
            for method_name, display_name in JUDGE.items():
                self._method_list.append({"method_name": method_name, "display_name": display_name})
        return self._method_list

    def get_value_operation_dict(self):
        """
        获取支持的数据转换方法字典，格式为{数据类型：支持的数据转换方法列表，...}
        数据转换方法列表是元素为方法描述字典的列表
        方法描述字典格式为 {"method_name": method_name, "display_name": display_name}
        如：{"method_name": "zfill", "display_name": "头部用0填充到指定长度"}
        :return:
        """
        if not self._op_dict:
            op_list = list()
            for method_name, display_name in STR_OP.items():
                op_list.append({"method_name": method_name, "display_name": display_name})
            self._op_dict["text"] = copy.deepcopy(op_list)
            op_list = list()
            for method_name, display_name in DATETIME_OP.items():
                op_list.append({"method_name": method_name, "display_name": display_name})
            self._op_dict["datetime"] = copy.deepcopy(op_list)
            op_list = list()
            for method_name, display_name in NUM_OP.items():
                op_list.append({"method_name": method_name, "display_name": display_name})
            self._op_dict["default"] = copy.deepcopy(op_list)
        return self._op_dict

    def filter_data(self, origin_df, filter_expr_list: list):
        """
        用于在Dataframe中根据过滤条件过滤出数据，返回符合条件的Dataframe
        :param origin_df: 原始数据框
        :param filter_expr_list: 过滤条件列表
            过滤条件格式为列表，形式为[过滤表达式, and或or连接符， 过滤表达式1, and或or连接符， 过滤表达式2...]
            过滤表达式格式为字典，形式为{filter_value：数值表达式，filter_op: 过滤判断方法字符串，
                                    op_value1：数值表达式， op_value2：数值表达式}
            过滤判断方法字符串支持的方法有：
                ['__contains__', '__eq__', '__ge__', '__gt__', '__le__', '__lt__', '__ne__', 'endswith', 'isdigit',
                  'startswith', 'in', 'between']
            数值表达式格式为字典， 形式为{const_value: 固定值， \
                            'asset': 资产表名（"api":/"app"/"account"/"host"), \
                            field_name: log_hour, trans: 转换方法字符串, args: 转换参数}
            转换方法字符串支持的方法有：
                STR_OP_LIST = ['__add__', '__len__', '__mul__', '__sizeof__', '__str__', 'lower', 'lstrip', 'replace',
                                'rstrip', 'strip', 'upper', 'zfill']
                NUM_OP_LIST = ['__abs__', '__add__', '__float__', '__floordiv__', '__mod__',
                                '__mul__', '__pow__', '__str__']
                DATETIME_OP_LIST = ['__str__', 'date', 'day', 'hour', 'microsecond', 'minute',
                                    'month', 'second','weekday', 'year']
            转换参数可以是数值或字符串，多个字符串以|连接作为一个参数

            For Example：
                过滤条件 [{"filter_value"：{"field_name": logtime, "trans": "hour", args: []}，"filter_op": "between"，
                        "op_value1"：{"const_value": 8}， "op_value2"：{"const_value": 20}}]
                这是一个单过滤条件
                意思为：取Dataframe的logtime字段，并对其执行hour方法，即：logtime的hour结果在固定值8和固定值20之间，
                        业务含义为过滤工作时间的日志，工作时间设置为8点到20点
        :return:
        """
        # 如果过滤条件为空、格式检查未通过、入参不是数据框格式或者数据为空，直接返回原值
        if (not filter_expr_list or not self._expr_list_check(filter_expr_list)
                or not isinstance(origin_df, pd.DataFrame) or not len(origin_df)):
            self.logger.debug(
                "[DataFrameFilter.filter_data]::过滤条件为空、格式检查未通过、非数据框格式或者数据为空，直接返回原值")
            return origin_df

        t = time.time()
        origin_len = len(origin_df)
        origin_df_columns = list(origin_df.columns)
        self.logger.debug("origin_df_columns:{origin_df_columns}")

        result_df_list = list()
        chunk_size = 100000
        for _start in range(0, len(origin_df), chunk_size):
            _end = _start + chunk_size
            chunk_df = origin_df.iloc[_start: _end]
            filter_df, result_field = self._calc_filter_lst_res(chunk_df, filter_expr_list)

            self.logger.debug(f"{filter_df.head(5)}")
            self.logger.debug(f"result_field:{result_field}")
            filter_df = filter_df[filter_df[result_field]]

            filter_df = filter_df.loc[:, origin_df_columns]
            self.logger.info(f"批次操作结果：从{_end - _start}条数据中过滤出{len(filter_df)}条数据")
            if len(filter_df):
                self.logger.debug(f"filter_expr_list:{filter_expr_list}")

                self.logger.debug("after filter, head 5 of df is:")
                for row in filter_df.head(5).itertuples():
                    self.logger.debug(f"{row}")

                result_df_list.append(filter_df)
        if not len(result_df_list):
            return pd.DataFrame(columns=origin_df_columns)
        if len(result_df_list) == 1:
            filter_df = result_df_list[0]
        else:
            filter_df = pd.concat(result_df_list, axis=0)
        self.logger.debug(f"[DataFrameFilter.filter_data]::过滤数据耗时{time.time() - t}秒")
        self.logger.info(
            f"[DataFrameFilter.filter_data]::过滤数据完成，过滤出{len(filter_df)}条数据，原有数据{origin_len}条")

        del origin_df
        gc.collect()

        return filter_df

    def _data_transfer(self, value, trans_op: str, trans_args: list = None):
        """
        根据转换方法和转换参数对数据进行转换
        转换方法字符串支持的方法有：
                STR_OP_LIST = ['__add__', '__len__', '__mul__', '__sizeof__', '__str__', 'lower', 'lstrip', 'replace',
                                'rstrip', 'strip', 'upper', 'zfill']
                NUM_OP_LIST = ['__abs__', '__add__', '__float__', '__floordiv__', '__mod__',
                                '__mul__', '__pow__', '__str__']
                DATETIME_OP_LIST = ['__str__', 'date', 'day', 'hour', 'microsecond', 'minute',
                                    'month', 'second','weekday', 'year']
        使用实例：
            BaseFilter._data_transfer(string, "__len__", [])  --> 将 string 转换为其长度
            BaseFilter._data_transfer(string, "zfill", [10])  --> 将 string 在头部补0为一个长度为10的字符串
            BaseFilter._data_transfer(number, "__pow__", [2])  --> 将 number 转换为其二次方
            BaseFilter._data_transfer(datetime, "year", [])  --> 将 datetime 转换为其年份数字
        :param value: 待转换的数据
        :param trans_op: 转换方法
        :param trans_args: 转换参数
        :return:
        """
        if not trans_op:
            return value
        _trans = getattr(value, trans_op)
        if hasattr(_trans, "__call__"):
            if trans_args:
                try:
                    return _trans(*trans_args)
                except TypeError as e:
                    self.logger.warning(
                        f"数据转换遇到格式错误，尝试使用不带参数，数据为：{value}，数据类型为：{type(value)}, "
                        f"转换方法为：{trans_op}, 参数为：{trans_args}, 错误信息：{e}")
                    return _trans()
            else:
                return _trans()
        else:
            return _trans

    def _calc_filter_lst_res(self, origin_df, filter_expr_list):
        """
        根据过滤条件，计算过滤结果，将结果（布尔型）作为新列添加到原数据框中，返回带结果的数据框以及结果列名
        :param origin_df:
        :param filter_expr_list:
        :return:
        """
        result_df = origin_df
        old_columns = list(result_df.columns)
        result_field_list = list()
        join_list = list()
        for i in range(len(filter_expr_list)):
            _expr = filter_expr_list[i]
            if isinstance(_expr, list):
                result_df, result_field = self._calc_filter_lst_res(result_df, _expr)
                result_field_list.append(result_field)
            elif isinstance(_expr, str):
                join_list.append(_expr)
            elif isinstance(_expr, dict):
                result_df, result_field = self._calc_filter_res(result_df, _expr)
                result_field_list.append(result_field)
            else:
                raise SyntaxError("过滤表达式非法，过滤表达式的元素只能是列表、字典或and/or字符串")

        result_field = result_field_list[0]
        result_index = 1
        for join_op in join_list:
            result_field2 = result_field_list[result_index]
            result_index += 1
            new_field_name = str(uuid.uuid4())
            if join_op == "and":
                result_df[new_field_name] = result_df[result_field] & result_df[result_field2]
            elif join_op == "or":
                result_df[new_field_name] = result_df[result_field] | result_df[result_field2]
            else:
                raise SyntaxError(f"格式非法！！！条件组合只支持 and 或 or，现在给的是： '{join_op}'")
            result_field = new_field_name
        old_columns.append(result_field)
        result_df = result_df.loc[:, old_columns]

        return result_df, result_field

    def _calc_filter_res(self, result_df: pd.DataFrame, filter_expr):
        """
        对每一个单独的表达式，富化转换后的字段，然后根据过滤条件，生成过滤结果列（True或者False）
        返回包含过滤结果列的新数据框和过滤结果列名
        :param result_df:
        :param filter_expr:
        :return:
        """
        old_columns = list(result_df.columns)
        new_expr = copy.deepcopy(filter_expr)
        filter_value: dict = filter_expr.get("filter_value")
        result_df, new_field_name = self._enrich_value_trans_field(result_df, filter_value)
        if new_field_name:
            value_expr = {"field_name": new_field_name}
            new_expr["filter_value"] = value_expr
        op_value1: dict = filter_expr.get("op_value1", dict())
        result_df, new_field_name = self._enrich_value_trans_field(result_df, op_value1)
        if new_field_name:
            value_expr = {"field_name": new_field_name}
            new_expr["op_value1"] = value_expr
        op_value2: dict = filter_expr.get("op_value2", dict())
        result_df, new_field_name = self._enrich_value_trans_field(result_df, op_value2)
        if new_field_name:
            value_expr = {"field_name": new_field_name}
            new_expr["op_value2"] = value_expr
        result_df, result_field = self._enrich_expr_result(result_df, new_expr)
        old_columns.append(result_field)
        result_df = result_df.loc[:, old_columns]
        return result_df, result_field

    @staticmethod
    def _is_column_ip_type(field_name: str, df: pd.DataFrame):
        """
        判断数据框列是否是ip地址列，返回0表示不是，返回1表示是整型IP，返回2表示是字符串型IP
        :param field_name:
        :param df:
        :return:
        """
        _dtype = str(df[field_name].dtype)
        if 'int' in _dtype and 'ip' in field_name.lower():
            return 1
        if 'object' in _dtype and 'ip' in field_name.lower():
            return 2
        return 0

    def _ip_to_int(self, ip_str):
        try:
            packed = socket.inet_aton(ip_str)
            res = struct.unpack("!I", packed)
            if res:
                return res[0]
            else:
                return 4294967295  # 255.255.255.255的整型表示
        except Exception as e:
            self.logger.error(f"IP字符串转整型出错，参数为：{ip_str},错误信息为：{e}")
            return 4294967295  # 255.255.255.255的整型表示

    def _str_arg_to_int_or_datetime(self, op_args: list, dst_type: str = 'ip'):
        new_args = list()
        if dst_type == 'ip':
            trans_func = self._ip_to_int
        elif dst_type == 'datetime':
            trans_func = self._convert_param
        else:
            return op_args

        for _arg in op_args:
            if isinstance(_arg, str):
                new_arg = trans_func(_arg)
            elif isinstance(_arg, list):
                new_arg = [trans_func(i) for i in _arg]
            else:
                new_arg = _arg
            new_args.append(new_arg)
        return new_args

    def _convert_param(self, param) -> pd.Timestamp:
        """统一转换查询时间参数"""
        default_res = pd.to_datetime("1970-01-01", format='mixed')
        try:
            if isinstance(param, pd.Timestamp):
                return param.tz_localize(None)
            elif isinstance(param, datetime):
                return pd.to_datetime(param)
            elif isinstance(param, (int, float)):
                # 根据数值长度自动判断单位
                param = int(param)
                if 1e8 <= param < 1e10:
                    return pd.to_datetime(param, unit='s')
                elif 1e12 <= param < 1e14:
                    return pd.to_datetime(param, unit='ms')
                else:
                    self.logger.error(f"参数转时间转换失败 | 非常规时间戳长度: {len(str(param))}位")
                    return default_res
            elif isinstance(param, str):
                return pd.to_datetime(param, format='mixed')
            else:
                self.logger.error(f"参数转时间转换失败 | 不支持的参数类型: {type(param)}")
                return default_res
        except Exception as e:
            self.logger.error(f"参数转时间转换失败 | param:{param} | {str(e)}")
            return default_res

    @staticmethod
    def _special_op_change(filter_op):
        if filter_op == "in":
            return "isin"
        elif filter_op == "__contains__":
            return "contains"
        elif filter_op == "__len__":
            return "len"
        else:
            return filter_op

    def _enrich_expr_result(self, origin_df: pd.DataFrame, filter_expr):
        self.logger.debug(f"filter_expr:{filter_expr}")
        filter_value_expr = filter_expr.get("filter_value")
        field_name = filter_value_expr.get("field_name")
        filter_op: str = filter_expr.get("filter_op")
        op_value1_expr = filter_expr.get("op_value1", dict())
        op_value2_expr = filter_expr.get("op_value2", dict())
        op_args = list()
        value1 = self._get_arg_value(origin_df, op_value1_expr)
        if value1 is not None:
            op_args.append(value1)
        value2 = self._get_arg_value(origin_df, op_value2_expr)
        if value2 is not None:
            op_args.append(value2)
        self.logger.debug(f"field_name:{field_name}, filter_op:{filter_op}, args:{op_args}")

        # 如果操作列是整型IP，则将参数也转为整型IP，因为输入的条件默认都会是字符串IP，没有人会在配置时填整型IP
        if self._is_column_ip_type(field_name, origin_df) == 1:
            op_args = self._str_arg_to_int_or_datetime(op_args, dst_type='ip')

        # 如果操作列是时间格式，则将判断的参数也转为时间格式，因为输入条件默认都是字符串时间，没有人在配置时填时间戳或其他形式
        if pd.api.types.is_datetime64_any_dtype(origin_df[field_name]):
            op_args = self._str_arg_to_int_or_datetime(op_args, dst_type='datetime')

        result_field_name = str(uuid.uuid4())
        # 先对 ['in', 'not_in', 'not_between']做特殊处理
        # 如果是not_开头的，去掉not_，在正常做in和between操作后，最后对结果取反
        # 如果是in，则改为isin，因为pandas用isin定义的in操作
        _reverse = False
        if filter_op.startswith("not_"):
            _reverse = True
            filter_op = filter_op[4:]

        filter_op = self._special_op_change(filter_op)

        _func = self._get_columns_op_func_obj(origin_df, field_name, filter_op)

        origin_df = origin_df.copy()
        origin_df[result_field_name] = _func(*op_args)
        if _reverse:
            origin_df[result_field_name] = ~origin_df[result_field_name]

        self.logger.debug(f"filter_expr:{filter_expr}")
        self.logger.debug("after filter, head 5 of df is:")
        for row in origin_df.head(5).itertuples():
            self.logger.debug(f"{row}")

        return origin_df, result_field_name

    def _get_arg_value(self, origin_df: pd.DataFrame, value_expr):
        const_value = value_expr.get("const_value", None)
        if const_value is not None:
            if isinstance(const_value, str) and "|" in const_value:
                return const_value.split("|")
            else:
                return const_value
        value_field = value_expr.get("field_name", "")
        if value_field:
            return origin_df[value_field]
        ast_name = value_expr.get("asset", "")
        if ast_name and hasattr(self.asset_cls, "get_ast"):
            return self.asset_cls.get_ast(ast_name)
        return None

    def _get_columns_op_func_obj(self, origin_df: pd.DataFrame, field_name: str, func_name: str):
        # pandas的一列是一个Series对象，由于Series本身和字符串Series的字符串值都支持replace方法，而我们想要的是值的replace操作
        # 所以针对replace做特殊处理，直接取.str.replace()
        if "replace" in func_name:
            return getattr(origin_df[field_name].str, func_name)

        _func, res, _ = self._try_get_columns_op_func_obj(origin_df, field_name, func_name)
        if res:
            return _func
        else:
            _func, res, err_msg = self._try_get_columns_op_func_obj(origin_df, field_name, func_name.strip('_'))
            if res:
                return _func
            else:
                raise AttributeError(err_msg)

    @staticmethod
    def _try_get_columns_op_func_obj(origin_df: pd.DataFrame, field_name: str, func_name: str):
        try:
            _func = getattr(origin_df[field_name], func_name)
        except AttributeError as e:
            try:
                if pd.api.types.is_string_dtype(origin_df[field_name]):
                    _func = getattr(origin_df[field_name].str, func_name)
                elif pd.api.types.is_datetime64_any_dtype(origin_df[field_name]):
                    _func = getattr(origin_df[field_name].dt, func_name)
                else:
                    return None, False, str(e)
            except AttributeError as e:
                return None, False, str(e)
        return _func, True, ""

    def _enrich_value_trans_field(self, data_df: pd.DataFrame, value_expr: dict):
        """
        根据数据转换的操作富化出相应的临时列，后续使用临时列进行列向量操作以提高数据筛选效率
        如： 数值表达式为：expr = {"field_name": "ID", "trans_op": "__add__", "args": [1]}
            经过此函数处理后，会在原数据框中生成新的一列，列名通过UUID4随机生成，值就是按照表达式转换出的值
        :param data_df:
        :param value_expr:
        :return:
        """
        if "trans" not in value_expr:
            self.logger.debug("无需富化")
            return data_df, ""
        # 从数据框的一行数据中解析出需要的字段值，如需转换，则返回转换后的值
        field_name = value_expr.get("field_name")
        trans = value_expr.get("trans")
        args = value_expr.get("args", list())
        new_field_name = str(uuid.uuid4())

        # __str__方法用astype特殊处理
        if trans == "__str__":
            data_df = data_df.copy()
            data_df[new_field_name] = data_df[field_name].astype(str)
            return data_df, new_field_name
        trans = self._special_op_change(trans)
        _trans = self._get_columns_op_func_obj(data_df, field_name, trans)

        data_df = data_df.copy()
        if hasattr(_trans, "__call__"):
            # 增加参数类型保护，主要是针对纯数字被当作字符串传入的情况
            try:
                data_df[new_field_name] = _trans(*args)
            except TypeError:
                new_args = list()
                for _arg in args:
                    if isinstance(_arg, str) and _arg.isdigit():
                        new_args.append(int(_arg))
                    else:
                        new_args.append(_arg)
                data_df[new_field_name] = _trans(*new_args)
        else:
            data_df[new_field_name] = _trans

        return data_df, new_field_name

    def _expr_list_check(self, expr_list):
        if not expr_list:
            return False
        try:
            json.dumps(expr_list, ensure_ascii=False)
        except Exception as e:
            self.logger.warning(f"过滤条件表达式列表格式无效，不是有效的json结构, 错误信息：{e}")
            return False
        if not isinstance(expr_list, list):
            self.logger.warning("过滤条件表达式列表格式无效，不是列表")
            return False
        true_expr_list = self._parse_expr_from_expr_list(expr_list)
        link_symbol_list = self._parse_link_symbol_from_expr_list(expr_list)

        if not self._expr_check(true_expr_list):
            return False
        if not self._link_symbol_check(link_symbol_list):
            return False
        return True

    @staticmethod
    def _parse_expr_from_expr_list(expr_list):
        return [expr_list[i] for i in range(len(expr_list)) if i % 2 == 0]

    @staticmethod
    def _parse_link_symbol_from_expr_list(expr_list):
        return [expr_list[i] for i in range(len(expr_list)) if i % 2 == 1]

    @capt_exc(_return=False, logger_name=error_subject_name, extra_msg="表达式检查时出现错误")
    def _expr_check(self, _expr_list: list):
        for _expr in _expr_list:
            if isinstance(_expr, list):
                return self._expr_list_check(_expr)
            if not isinstance(_expr, dict):
                self.logger.warning(f"过滤条件表达式格式无效，不是字典格式，表达式：{_expr}")
                return False
            if len(set(EXPR_MUST_HAS_KEYS & _expr.keys())) < 3:
                self.logger.warning(
                    f"过滤条件表达式格式无效，缺少必填字段，必填字段为：{EXPR_MUST_HAS_KEYS}，表达式：{_expr}")
                return False
            if "filter_op" not in _expr or _expr["filter_op"] not in JUDGE:
                self.logger.warning(
                    f"过滤判断方法字段 'filter_op'不存在或判断方法不存在, 支持的判断方法包括：{JUDGE}，表达式：{_expr}")
                return False
            op_value1 = _expr.get("op_value1", dict())
            filter_value = _expr.get("filter_value", dict())
            op_value2 = _expr.get("op_value2", dict())
            value_list = [op_value1, filter_value, op_value2]
            for value_expr in value_list:
                if not self._value_expr_check(value_expr):
                    return False
        return True

    def _value_expr_check(self, value_expr):
        if not isinstance(value_expr, dict):
            self.logger.warning(f"数值表达式不合法，不是字典类型，数值表达式：{value_expr}")
            return False
        if value_expr.get("asset", None) is not None and self.asset_cls is None:
            self.logger.warning("过滤条件用到了资产数据，但是这个类实例初始化时未给定有效的资产类")
            return False
        if value_expr.get("asset", None) is not None and value_expr.get("asset", None) not in self.asset_name_set:
            self.logger.warning(f"不支持的资产类别：{value_expr.get('asset')}, 支持的资产类别包括：{self.asset_name_set}")
            return False

        trans = value_expr.get("trans", None)
        if trans is not None and trans not in VALID_OP_SET:
            self.logger.warning(
                f"不支持的数据转换方法, 支持的方法包括：{VALID_OP_SET}，数值表达式：{value_expr}")
            return False
        return True

    def _link_symbol_check(self, link_symbol_list):
        try:
            for link_symbol in link_symbol_list:
                if not isinstance(link_symbol, str) or ("and" not in link_symbol and "or" not in link_symbol):
                    self.logger.warning(
                        f"过滤条件表达式连接符不合法，必须是字符串，且支持 and 和 or，输入的连接符为：{link_symbol}")
                    return False
            return True
        except Exception as e:
            self.logger.warning(f"连接符检查时出现错误，错误信息：{e}")
            return False
